package org.fujene.communicate.searcher;

import java.util.Arrays;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.Map.Entry;
import java.util.concurrent.LinkedBlockingQueue;
import org.fujene.communicate.hash.Hashing;
import org.fujene.communicate.node.Node;
import org.fujene.communicate.rpc.Communicator;
import org.fujene.logging.LogParser;
import org.fujene.memman.MemoryManager;
import org.fujene.memman.TermInformation;
import org.fujene.memoryblock.dictionary.DictionaryPackage;

public class TermSearcherHost implements Runnable{
	private static SortedSet<Node> nodes;
	private static MemoryManager memman;
	private static Node myNode;
	private SearcherPackage structure;
	private String term;
	private int count;
	private byte[] contentTypes;
	private static int replication;
	private static LinkedBlockingQueue<SearcherPackage> queue;
	private static Communicator communicator;
	private int retryCount = 2;

	static{
		queue = new LinkedBlockingQueue<SearcherPackage>();
	}

	public TermSearcherHost(SearcherPackage structure, int count, byte[] contentTypes){
		this.term = structure.getTerm(count).str();
		this.structure = structure;
		this.contentTypes = contentTypes;
		this.count = count;
	}

	public static void property(int repl, MemoryManager manager, Node ownNode,
		SortedSet<Node> list, Communicator comm){
		replication = repl;
		memman = manager;
		myNode = ownNode;
		nodes = list;
		communicator = comm;
	}

	public static void addToQueue(SearcherPackage newSearch){
		queue.add(newSearch);
	}

	public static int residue(){
		return queue.size();
	}

	public void run(){
		try{
			long hash = Hashing.getTermHash(term);
			TreeMap<Long, Node> hashList = new TreeMap<Long, Node>();
			Entry<Long, Node> currentEntry;
			Node termNode = null;
			for(Node eachNode: nodes)
				hashList.put(eachNode.getAttributeHash(hash), eachNode);
			currentEntry = hashList.ceilingEntry(Long.valueOf(hash));
			if(currentEntry == null) currentEntry = hashList.firstEntry();
			int cluster = nodes.headSet(currentEntry.getValue()).size();
//			System.out.println("[DEBUG]Search queue - Cluster: " + cluster);

			// search boosting
			termNode = currentEntry.getValue();
			for(int i = 0; i < replication; i++){
				if(currentEntry.getValue() == myNode){
					termNode = currentEntry.getValue();
					break;
				}
//				termNode = currentEntry.getValue();
				currentEntry = hashList.lowerEntry(currentEntry.getKey());
				if(currentEntry == null) currentEntry = hashList.lastEntry();
			}

			LogParser.fine("Search - Node: " + termNode);
			if(termNode == myNode){
				// Search locally
				TermInformation info = memman.searchTerm(term, contentTypes, cluster);
				DictionaryPackage pkg = info.getDictionary();
				if(pkg == null){
					structure.notFound();
					return;
				}
				for(int i = 0; i < contentTypes.length; i++){
					for(int j = 0; j < 8; j++)
						if((contentTypes[i] & (1 << j)) != 0)
							structure.setResult(memman.getInvertIndex(
								info.getDictionary().getInvertIndex(8 * i + j), structure.getFilter()));
				}
			}else{
				long[][] remoteInvIndex = null;
				for(int i = 0; i < retryCount; i++){
					try{
						long time = LogParser.initTime();
						remoteInvIndex =
							communicator.searchRemote(term, contentTypes, termNode, cluster,
								structure.getFilter());
						LogParser.timerFinest("Search / Post to remote", time);
						for(int j = 0; j < remoteInvIndex.length; j++)
							structure.setResult(remoteInvIndex[j]);
						break;
					}catch(Exception e){
						LogParser.warning("Retrying search on next node.");
						currentEntry = hashList.lowerEntry(currentEntry.getKey());
						if(currentEntry == null) currentEntry = hashList.lastEntry();
						continue;
					}
				}
			}
			// all searches demolish if result's NULL
			if(structure.getResult(count) == null) structure.notFound();
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			structure.countdown();
		}
	}
}
